package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.DwsTestExamWindonBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTestExamWindon {
    public static void main(String[] args) throws Exception {
        // TODO 1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO  2、设置检查点和状态后端
        /*env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        //延迟时间
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
        //最多可以出现多少状态后端
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //检查点保存位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //开启状态后端
        env.setStateBackend(new HashMapStateBackend());*/

        //TODO 3、读取Kafka数据
        String topicName = "dwd_text_exam";
        String groupId = "dwd_test_exam_windon";
        DataStreamSource<String> kafkaStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //TODO 4、转换数据格式
        SingleOutputStreamOperator<JSONObject> mapStream = kafkaStream.map(JSONObject::parseObject);
        //TODO 5、keyby键控状态
        KeyedStream<JSONObject, String> keyedStream = mapStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("paper_id")+value.getString("user_id");
            }
        });
        //TODO 6、对同一试卷id的uid进行去重
        SingleOutputStreamOperator<DwsTestExamWindonBean> flatMapStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, DwsTestExamWindonBean>() {
            ValueState<String> pid_uid = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                pid_uid = getRuntimeContext().getState(new ValueStateDescriptor<String>("pid_uid", String.class));
            }

            @Override
            public void flatMap(JSONObject value, Collector<DwsTestExamWindonBean> out) throws Exception {
                String value1 = pid_uid.value();
                String key = value.getString("paper_id")+value.getString("user_id");
                if (!key.equals(value1)) {
                    out.collect(DwsTestExamWindonBean.builder()
                            .paper_id(value.getString("paper_id"))
                            .during_time_sum(value.getLong("duration_sec"))
                            .user_cnt(1L)
                            .score_sum(value.getLong("score"))
                            .ts(DateFormatUtil.toTs(value.getString("submit_time"),true)).build());
                    pid_uid.update(value.getString("user_id"));
                } else {
                }
            }
        });
        //flatMapStream.print("flatmap");

        SingleOutputStreamOperator<DwsTestExamWindonBean> watermarkStream = flatMapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsTestExamWindonBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<DwsTestExamWindonBean>() {
            @Override
            public long extractTimestamp(DwsTestExamWindonBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        KeyedStream<DwsTestExamWindonBean, String> keyedStream1 = watermarkStream.keyBy(new KeySelector<DwsTestExamWindonBean, String>() {
            @Override
            public String getKey(DwsTestExamWindonBean value) throws Exception {
                return value.getPaper_id();
            }
        });
        SingleOutputStreamOperator<DwsTestExamWindonBean> reduce = keyedStream1
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsTestExamWindonBean>() {
                    @Override
                    public DwsTestExamWindonBean reduce(DwsTestExamWindonBean value1, DwsTestExamWindonBean value2) throws Exception {
                        value1.setUser_cnt(value1.getUser_cnt() + value2.getUser_cnt());
                        value1.setDuring_time_sum(value1.getDuring_time_sum() + value2.getDuring_time_sum());
                        value1.setScore_sum(value1.getScore_sum() + value2.getScore_sum());
                        return value1;
                    }
                }, new WindowFunction<DwsTestExamWindonBean, DwsTestExamWindonBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DwsTestExamWindonBean> input, Collector<DwsTestExamWindonBean> out) throws Exception {
                        DwsTestExamWindonBean next = input.iterator().next();

                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setSystime(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        reduce.print("dddddddddddddddd");


        SingleOutputStreamOperator<DwsTestExamWindonBean> course_id = AsyncDataStream.unorderedWait(reduce, new DimAsyncFunction<DwsTestExamWindonBean>("DIM_TEST_PAPER") {
            @Override
            public String getKey(DwsTestExamWindonBean input) {
                return input.getPaper_id();
            }

            @Override
            public void join(DwsTestExamWindonBean input, JSONObject obj) {
                input.setCourse_id(obj.getString("courseId"));
            }
        }, 100, TimeUnit.SECONDS);

        SingleOutputStreamOperator<DwsTestExamWindonBean> course_name = AsyncDataStream.unorderedWait(course_id, new DimAsyncFunction<DwsTestExamWindonBean>("DIM_COURSE_INFO") {
            @Override
            public String getKey(DwsTestExamWindonBean input) {
                return input.getCourse_id();
            }

            @Override
            public void join(DwsTestExamWindonBean input, JSONObject obj) {
                input.setCourse_name(obj.getString("courseName"));

            }
        }, 100, TimeUnit.SECONDS);

        course_name.print("ssssss");

        env.execute(groupId);

        //
    }
}
